package com.study.flume;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 类型拦截器：将消息分发至不同kafka主题
 *
 * @author stephen
 * @date 2019-09-03 14:02
 */
public class TypeInterceptor  implements Interceptor {
    /**
     * 编码
     */
    String CHARSET_NAME = "UTF-8";

    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        // 获取数据
        byte[] body = event.getBody();
        String log = new String(body, Charset.forName(CHARSET_NAME));
        // 获取header
        Map<String, String> headers = event.getHeaders();
        // 判断日志中是否包含hello，包含则发送至hello topic，否则发送至test topic
        if (log.contains("hello")) {
            headers.put("topic", "hello");
        } else {
            headers.put("topic", "test");
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        ArrayList<Event> interceptors = new ArrayList<>();
        for (Event event : events) {
            Event intercept = intercept(event);
            interceptors.add(intercept);
        }
        return interceptors;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new TypeInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }

}
